package com.steakliu.sun.codec;

import com.steakliu.sun.common.thread.FlowControlPostProcessorThreadPool;
import com.steakliu.sun.common.util.CodecUtil;
import com.steakliu.sun.constant.SunConstant;
import com.steakliu.sun.flow.api.FlowControlPostProcessor;
import com.steakliu.sun.protocol.*;
import com.steakliu.sun.serializer.api.SunSerializer;
import com.steakliu.sun.spi.loader.ExtensionLoader;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.springframework.context.event.ContextRefreshedEvent;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 功能说明： 解码器
 * <p>
 * Original @Author: steakliu-刘牌, 2023-01-08  12:42
 * <p>
 * Copyright (C)2020-2022  steakliu All rights reserved.
 */
public class SunDecoder extends ByteToMessageDecoder {

    private FlowControlPostProcessor flowControlPostProcessor;

    private final boolean isUseFlowControl;

    public SunDecoder(boolean isUseFlowControl, String flowControlType) {
        this.isUseFlowControl = isUseFlowControl;
        if (isUseFlowControl) {
            this.flowControlPostProcessor = ExtensionLoader.getExtension(FlowControlPostProcessor.class, flowControlType);
        }
    }

    @Override
    protected void decode(ChannelHandlerContext context, ByteBuf byteBuf, List<Object> list) throws Exception {
        if (byteBuf.readableBytes() < SunConstant.MSG_MIN_LEN) {
            return;
        }
        byteBuf.markReaderIndex();
        short magic = byteBuf.readShort();
        if (magic != SunConstant.MAGIC) {
            throw new RuntimeException("The magic number is incorrect, and the message is illegal!");
        }
        byte type = byteBuf.readByte();
        byte status = byteBuf.readByte();
        long requestId = byteBuf.readLong();
        ByteBuf serializerByteBuf = byteBuf.readBytes(CodecUtil.SERIALIZER_MAX_LEN);
        String serializerName = CodecUtil.getSerializerStr(serializerByteBuf.toString(StandardCharsets.UTF_8));
        int msgLen = byteBuf.readInt();
        if (byteBuf.readableBytes() < msgLen) {
            byteBuf.resetReaderIndex();
            return;
        }
        byte[] messageBodyBytes = new byte[msgLen];
        byteBuf.readBytes(messageBodyBytes);//消息体

        MsgType msgType = MsgType.findType(type);
        if (null == msgType) {
            return;
        }
        MsgHeader header = new MsgHeader();
        header.setMagic(magic);
        header.setStatus(status);
        header.setType(type);
        header.setRequestId(requestId);
        header.setSerializer(serializerName);
        header.setMsgLen(msgLen);
        SunSerializer serializer = ExtensionLoader.getExtension(SunSerializer.class, serializerName);
        switch (msgType) {
            case REQUEST:
            case CLIENT_HEARTBEAT:
                SunProtocol<SunRequest> requestSunProtocol = new SunProtocol<>();
                SunRequest request = serializer.deserialize(messageBodyBytes, SunRequest.class);
                if (null != request) {
                    requestSunProtocol.setMsgHeader(header);
                    requestSunProtocol.setBody(request);
                    list.add(requestSunProtocol);
                }
                break;
            case RESPONSE:
            case SERVER_RESPONSE_HEARTBEAT:
                SunProtocol<SunResponse> responseSunProtocol = new SunProtocol<>();
                SunResponse response = serializer.deserialize(messageBodyBytes, SunResponse.class);
                if (null != response) {
                    responseSunProtocol.setMsgHeader(header);
                    responseSunProtocol.setBody(response);
                    list.add(responseSunProtocol);
                }
                break;
        }
        //流控切入
        if (this.isUseFlowControl) {
            FlowControlPostProcessorThreadPool.submit(() -> {
                flowControlPostProcessor.postProcess(header);
            });
        }
    }
}
